package org.geekbang.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;

public class HbaseTest {

    public static void main(String[] args) throws IOException {
        // 建立连接
        Configuration configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.quorum", "emr-worker-2,emr-worker-1,emr-header-1");
        configuration.set("hbase.zookeeper.property.clientPort", "2181");
        Connection conn = ConnectionFactory.createConnection(configuration);
        Admin admin = conn.getAdmin();

        TableName tableName = TableName.valueOf("student");
        String colFamily1 = "name";
        String colFamily2 = "info";
        String colFamily3 = "score";

        int rowKey1 = 1;
        int rowKey2 = 2;
        int rowKey3 = 3;
        int rowKey4 = 4;

        // 建表
        if (admin.tableExists(tableName)) {
            System.out.println("表已经存在");
        } else {
            HTableDescriptor hTableDescriptor = new HTableDescriptor(tableName);
            HColumnDescriptor hColumnDescriptor1 = new HColumnDescriptor(colFamily1);
            HColumnDescriptor hColumnDescriptor2 = new HColumnDescriptor(colFamily2);
            HColumnDescriptor hColumnDescriptor3 = new HColumnDescriptor(colFamily3);
            hTableDescriptor.addFamily(hColumnDescriptor1);
            hTableDescriptor.addFamily(hColumnDescriptor2);
            hTableDescriptor.addFamily(hColumnDescriptor3);
            admin.createTable(hTableDescriptor);
            System.out.println("表创建成功");
        }

        // 插入第1行数据
        Put put1 = new Put(Bytes.toBytes(rowKey1)); // row key
        put1.addColumn(Bytes.toBytes(colFamily1), Bytes.toBytes("name"), Bytes.toBytes("Tom")); // col1
        put1.addColumn(Bytes.toBytes(colFamily2), Bytes.toBytes("student_id"), Bytes.toBytes("20210000000001")); // col2
        put1.addColumn(Bytes.toBytes(colFamily2), Bytes.toBytes("class"), Bytes.toBytes("1")); // col3
        put1.addColumn(Bytes.toBytes(colFamily3), Bytes.toBytes("understanding"), Bytes.toBytes("75")); // col4
        put1.addColumn(Bytes.toBytes(colFamily3), Bytes.toBytes("programming"), Bytes.toBytes("82")); // col5
        conn.getTable(tableName).put(put1);

        // 插入第2行数据
        Put put2 = new Put(Bytes.toBytes(rowKey2)); // row key
        put2.addColumn(Bytes.toBytes(colFamily1), Bytes.toBytes("name"), Bytes.toBytes("Jerry")); // col1
        put2.addColumn(Bytes.toBytes(colFamily2), Bytes.toBytes("student_id"), Bytes.toBytes("20210000000002")); // col2
        put2.addColumn(Bytes.toBytes(colFamily2), Bytes.toBytes("class"), Bytes.toBytes("1")); // col3
        put2.addColumn(Bytes.toBytes(colFamily3), Bytes.toBytes("understanding"), Bytes.toBytes("85")); // col4
        put2.addColumn(Bytes.toBytes(colFamily3), Bytes.toBytes("programming"), Bytes.toBytes("67")); // col5
        conn.getTable(tableName).put(put2);
        // 插入第3行数据
        Put put3 = new Put(Bytes.toBytes(rowKey3)); // row key
        put3.addColumn(Bytes.toBytes(colFamily1), Bytes.toBytes("name"), Bytes.toBytes("Jack")); // col1
        put3.addColumn(Bytes.toBytes(colFamily2), Bytes.toBytes("student_id"), Bytes.toBytes("20210000000003")); // col2
        put3.addColumn(Bytes.toBytes(colFamily2), Bytes.toBytes("class"), Bytes.toBytes("2")); // col3
        put3.addColumn(Bytes.toBytes(colFamily3), Bytes.toBytes("understanding"), Bytes.toBytes("80")); // col4
        put3.addColumn(Bytes.toBytes(colFamily3), Bytes.toBytes("programming"), Bytes.toBytes("80")); // col5
        conn.getTable(tableName).put(put3);
        // 插入第4行数据
        Put put4 = new Put(Bytes.toBytes(rowKey4)); // row key
        put4.addColumn(Bytes.toBytes(colFamily1), Bytes.toBytes("name"), Bytes.toBytes("Rose")); // col1
        put4.addColumn(Bytes.toBytes(colFamily2), Bytes.toBytes("student_id"), Bytes.toBytes("20210000000004")); // col2
        put4.addColumn(Bytes.toBytes(colFamily2), Bytes.toBytes("class"), Bytes.toBytes("2")); // col3
        put4.addColumn(Bytes.toBytes(colFamily3), Bytes.toBytes("understanding"), Bytes.toBytes("60")); // col4
        put4.addColumn(Bytes.toBytes(colFamily3), Bytes.toBytes("programming"), Bytes.toBytes("61")); // col5
        conn.getTable(tableName).put(put4);

        System.out.println("数据插入成功");

        // 查看数据
        Table table = conn.getTable(TableName.valueOf(String.valueOf(tableName)));
        Scan scan = new Scan();
        ResultScanner scanner = table.getScanner(scan);
        Result tmp;
        System.out.println("开始查询数据------------");
        while ((tmp = scanner.next()) != null) {
            List<Cell> cells = tmp.listCells();
            for (Cell cell : cells) {
                String colName = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
                String value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                System.out.println("列名: " + colName + ", 列值: " + value);
            }
        }
        System.out.println("查询数据结束------------");
        /*
        Get get = new Get(Bytes.toBytes(rowKey));
        if (!get.isCheckExistenceOnly()) {
            Result result = conn.getTable(tableName).get(get);
            for (Cell cell : result.rawCells()) {
                String colName = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
                String value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                System.out.println("数据查询成功, 列名: " + colName + ", 列值: " + value);
            }
        }
        */

        // 删除数据
        Delete delete = new Delete(Bytes.toBytes(rowKey2));      // 指定rowKey
        conn.getTable(tableName).delete(delete);
        System.out.println(tableName + " 第" + rowKey2 + "行 " + "数据删除成功" );

        // 删除表
        if (admin.tableExists(tableName)) {
            admin.disableTable(tableName);
            admin.deleteTable(tableName);
            System.out.println(tableName + "表删除成功");
        } else {
            System.out.println("表不存在！");
        }
    }
}


